package cz.cvut.fel.riak.loader;

import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.lang.RandomStringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.basho.riak.pbc.RequestMeta;
import com.basho.riak.pbc.RiakClient;
import com.basho.riak.pbc.RiakObject;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.Mongo;

import cz.cvut.fel.riak.email.Email;
import cz.cvut.fel.riak.parse.JavaMailParser;
import cz.cvut.fel.riak.parse.Parser;
import cz.cvut.fel.riak.persister.RabbitMQPersister;
import cz.cvut.fel.riak.stats.Statistics;

public class LoadJob extends Thread {
	public enum TEST_TYPE  { SEARCH, INSERT, UPDATE, DELETE};
	private TEST_TYPE testType = TEST_TYPE.SEARCH;
	
	final static Logger log = LoggerFactory.getLogger(LoadJob.class);
	private String path = null;
	private long iterations = 0;
	private String server = null;
	private RiakClient c;
	private Mongo mongo;
	private Parser parser = new JavaMailParser();
	private Statistics stats;
	public AtomicLong number;
	public boolean add;

	private RabbitMQPersister mq;
	private RiakPersister rp;
	public static byte[] data;
	private int batchSize;

	public LoadJob(String path, long iterations, String server, int batchSize) {
		super();
		this.path = path;
		this.iterations = iterations;
		this.server = server;
		stats = Statistics.getInstance();

		this.batchSize = batchSize;
		// /mq = new RabbitMQPersister("cvut2.centrum.cz", null);
		// rp = new RiakPersister(server, null);
		try {
			// rp.connect();
			// mq.connect();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void browse(String path) {
		File f = new File(path);
		if (f.isDirectory()) {
			for (int i = 0; i < f.listFiles().length; i++) {
				File ff = f.listFiles()[i];
				if (ff.isDirectory()) {
					browse(ff.getPath());

				} else {
					if (!ff.getPath().contains(".envelope")) {
						// log.info("To store "+ff.getPath());
						storeToRiak(parser.parseEmail(ff.getPath()));
						if ((iterations++ % 100) == 0)
							log.info("TH-ID: " + this.getId() + " WORKING:");// +stats.getStats()
																				// +
																				// "ops/s");
					}
				}
			}
		}
	}

	private void storeToRiak(Email email) {
		if (email == null) {
			log.info("ODSTRANIT");
			return;
		}

		for (int i = 0; i < email.getAttachments().size(); i++) {
			try {
				/*
				 * RiakObject obj = null; if (email.getAttachments().size() > 1
				 * && email.getAttachments().size() == (i-1)) obj = new
				 * RiakObject("TEST1",
				 * email.getKeys().get(i),email.getAttachments
				 * ().get(i).getBytes()); else obj = new
				 * RiakObject("ATTACHMENT1",
				 * email.getKeys().get(i),email.getAttachments
				 * ().get(i).getBytes());
				 * 
				 * c.store(obj);
				 */

				mq.store("null",	 email.getAttachments().get(i).getBytes());
				stats.add();
				// log.info("STORED "+email.getKeys().get(i));
			} catch (Exception e) {
				log.error("Unable to store email " + email, e);

			}
		}

		log.debug(" Thead-ID: " + this.getId() + " storing file " + path);

	}

	public void run() {
		log.info("Started thread with id: " + this.getId());
		int db = 1;
		try {
			String[] words = new String[1000];
			for (int z=0;z<1000;z++)
				words[z]=RandomStringUtils.random(((int)(Math.random()*10) % 10)+3, 0,0, true, false);
			
			if (server.contains(":")){
				c = new RiakClient(server.substring(0,server.lastIndexOf(":")),Integer.parseInt(server.substring(server.lastIndexOf(":")+1)));
				mongo = new Mongo(server.substring(0,server.lastIndexOf(":")),Integer.parseInt(server.substring(server.lastIndexOf(":")+1)));
		}else {
				c = new RiakClient(server);
				mongo = new Mongo(server);
			}
			DB dbase = mongo.getDB("BENCH");
			DBCollection coll = dbase.getCollection("col1");
			log.error("Tohle TODO se musi opravit");
			//m = new Mon
			// browse(this.path);
			RiakObject obj;
			BasicDBObject  bson;
			Random ran = new Random();
			String prefix = "" + ran.nextInt();
			
			batchSize = 1;
			long id = this.getId();
			id += 10;
			
			 HttpClient client = new HttpClient();
			 
			for (int i = 0; i < iterations; i++) {
				boolean success = false;
				long start = System.currentTimeMillis();
				
		/*		RiakObject[] objs = null;
				RequestMeta m = null ;
				if (db==0) {
				objs = new RiakObject[batchSize];				
				m= new RequestMeta();
				m.returnBody(false); // //////ZKUSIT ZNOVU*/
				//}
				//m.w(1);
				//m.dw(1);
				

				for (int r = 0; r < 10 && !success; r++) {
					try {
						  // Construct data
					    //String data = URLEncoder.encode("key1", "UTF-8") + "=" + URLEncoder.encode("value1", "UTF-8");
					    //data += "&" + URLEncoder.encode("key2", "UTF-8") + "=" + URLEncoder.encode("value2", "UTF-8");
						
					    // Send data
					    URL url = new URL("http://"+server+":8098/solr/DOMAIN/select?start=0&rows=0&q=DATA:"+RandomStringUtils.random(((int)(Math.random()*10) % 10)+3, 0,0, true, false));
					    
					    GetMethod met = new GetMethod(url.toString());
					    client.executeMethod(met);
					    
					    if (met.getStatusCode() != HttpStatus.SC_OK) {
					    	System.out.println("Returned CODE: "+met.getStatusCode() +"\nDATA: "+met.getResponseBodyAsString());
					    	continue;
					    } 
					    	//System.out.println(met.getResponseBodyAsString());
					    met.releaseConnection();
					    //System.out.println( met.getResponseBodyAsString());
					    //URLConnection conn = url.openConnection();
					    //conn.setDoOutput(true);
				//	    OutputStreamWriter wr = new OutputStreamWriter(conn.getOutputStream());
					    //wr.write(data);
					//    wr.flush();

					    // Get the response
/*					    conn.setDoInput(true);
					    BufferedReader rd = new BufferedReader(new InputStreamReader(conn.getInputStream()));
					    String line;
					    while ((line = rd.readLine()) != null) {
					        // Process line...
					    }*/
					    //wr.close();
					    //rd.close();
						
						/*						//prefix + iterations+ p
						if (batchSize > 1) {
							for (int p = 0; p < batchSize; p++) {
									objs[p] = new RiakObject("TESTING2", prefix + i + p, data);
							}
							

								c.store(objs, m);

								
						} else
						{
							if (db == 0) {
								obj = new RiakObject("TESTING2", prefix + i, data);
								c.store(obj, m);
							} else
							{
								//System.out.println(data.length);
								//bson = BSON.decode(data);
								
								bson = new BasicDBObject ();
								
								
								
								bson.put("_id",  (this.add ? this.number.getAndIncrement() : this.number.getAndDecrement()) );
								//bson.put("_id", UUID.randomUUID());
								
							//	bson.put("_id",  prefixIn+ (id*10000000)+i);
								//((ran.nextInt()*10000000)
								//bson.put("sKey", splitter[i%3]+""+ran.nextInt());
								//bson.put("_id", (""+splitter[i%3])+ ((this.getId() % 2)  == 1 ? Main.aiD.getAndIncrement() : Main.aiT.getAndDecrement()));  
										
										//prefixIn+Main.ai.getAndIncrement());
								//bson.put("_id", this.getId()+ ""+  i);
							//	bson.put("my", i%2);
								//bson.put(key, v) prefix + i
								bson.put("data", data);
								coll.insert(bson,WriteConcern.NORMAL);
						//		System.out.println( bson.get("_id"));
								//mongo.getDB("").getCollection("").insert(arr)
							
							}
						}*/
					    success = true;

					} catch (Exception e) {
						log.error("Restarting", e);
						c = new RiakClient(server);
					}
				}
				 stats.latency.getAndAdd(System.currentTimeMillis() - start);
				 int actual = stats.docs.getAndAdd(batchSize);
				 if (actual >= iterations) {
					 Main.remaining--;
					 return;
				 }
			}
			log.info("OK<<<<<<<<<<<<<<<<<<<<CLOSE");
			Main.remaining--;
			// String file = FileUtils.readFileToString(new File(path));
			// double x = Math.random();

		} catch (IOException e) {
			log.error("Erro in thead: " + this.getId() + "->" + e.getMessage(),
					e);
		}
	}

}
